/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.mqSender;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.unpack.file.global.GlobalParas;
import static cn.ac.iie.unpack.file.global.GlobalParas.MSGTYPES;
import cn.ac.iie.unpack.file.utils.FileListGetThread;
import cn.ac.iie.unpack.file.utils.FileUntarAndMoveThread;
import com.alibaba.fastjson.JSONObject;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import jdk.nashorn.internal.objects.NativeArray;
import org.apache.log4j.Logger;
import org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.commons.lang.exception.ExceptionUtils;

/**
 *
 * @author gjw
 */
public class DistriThread implements Callable<Integer> {

    static Logger LOG = Logger.getLogger(DistriThread.class);
    public static ArrayBlockingQueue<List<JSONObject>> msgQueue=new ArrayBlockingQueue(100);
    public static ExecutorService threadService=Executors.newFixedThreadPool(10);
    public static Map<String,SendMQThread>mqMap=new HashMap<String,SendMQThread>();
    public volatile static boolean Stop;
    public static List<Future> getDistriThread(){
        List<Future>  list=new ArrayList();
        for(int i=0;i<GlobalParas.MSGTYPES.length;i++){
            SendMQThread sendMq=new SendMQThread(GlobalParas.TOPICS[i], GlobalParas.MSGTYPES[i], GlobalParas.COLUMNS.get(i).toArray(new String[GlobalParas.COLUMNS.get(i).size()]),GlobalParas.TYPES.get(i).toArray(new String[GlobalParas.COLUMNS.get(i).size()]), GlobalParas.URLS.get(i),GlobalParas.DIRS[i]);
            mqMap.put(GlobalParas.MSGTYPES[i],sendMq);
            list.add(threadService.submit(sendMq));
        }
        return list;
    }

    @Override
    public Integer call() {
        List<JSONObject> dataList = null;
        long startTime = System.currentTimeMillis() / 1000;
        while (!FileUntarAndMoveThread.Stop||!msgQueue.isEmpty()) {
            try {
                dataList = msgQueue.poll(10, TimeUnit.SECONDS);
                if(dataList==null){
                    continue;
                }
                for(JSONObject json:dataList){
                    String type=json.getString("msg_type");
                    if(mqMap.get(type)==null){
                        continue;
                    }
                    mqMap.get(type).add(json);
                }
            } catch (InterruptedException ex) {
                LOG.error(ExceptionUtils.getFullStackTrace(ex));
            }catch (Exception ex) {
                LOG.error(ExceptionUtils.getFullStackTrace(ex));
            }
        }
        return 1;
    }
}
